package hbec.kettle.etl.engine.plugin;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.gson.Gson;

import hbec.kettle.etl.common.constants.DataBase;
import hbec.kettle.etl.plugins.common.DALUtils;
import hbec.kettle.etl.plugins.common.WatchLine;

public class BufferedLineExchanger {
	private static final Logger logger = LoggerFactory.getLogger(BufferedLineExchanger.class);
	
	private static Map<String, LinkedList<Line>> buffer = new ConcurrentHashMap<>();
	private static Map<Integer, Integer> failedBuffer = new ConcurrentHashMap<>();
	private static Map<String, Long> sqliteBuffer = new ConcurrentHashMap<>();
	private static Timer timer = new Timer();
	private static Gson gson = new Gson();
	
	public static synchronized boolean lineBuffer(String object, Line line) {
		boolean res = true;
		
		if (buffer.containsKey(object))
			if (buffer.get(object).size() < 100)
				buffer.get(object).add(line);
			else {
				String jsonLine = gson.toJson(line);
				int v = DALUtils.setSqlite(DataBase.SQLITE, object.split("\\.")[0].toUpperCase(), object, jsonLine);
				if (v <= 0) {
					res = false;
					logger.warn("sqlite insert fail");
				}
			}
		else {
			LinkedList<Line> linkedList = new LinkedList<>();
			linkedList.add(line);
			buffer.put(object, linkedList);
		}
		
		return res;
	}
	
	public static synchronized Line getLineBuffer(String object, Line line) {
		if (buffer.containsKey(object)) {
			if (buffer.get(object).size() == 0) {
				List<WatchLine> list = DALUtils.getSqlite(DataBase.SQLITE, object.split("\\.")[0].toUpperCase(), object, sqliteBuffer.get(object));
				if (list.size() > 0) {
					List<Line> lines = new ArrayList<>();
					Long id = 0L;
					for (WatchLine wl : list) {
						id = wl.getId();
						lines.add(gson.fromJson(wl.getLine(), Line.class));
					}
					sqliteBuffer.put(object, id);
					buffer.get(object).addAll(lines);
					return buffer.get(object).getFirst();
				} else 
					return line;
			} else {
				buffer.get(object).add(line);
				return buffer.get(object).getFirst();
			}
		} else 
			return line;
	}
	
	public static synchronized void removeLineBuffer(String object, Line line) {
		if (buffer.containsKey(object))
			buffer.get(object).remove(line);
	}
	
	public static synchronized void addFailedObject(final String object, final Line line) {
		int i = 0;
		if (failedBuffer.containsKey(Integer.valueOf(line.hashCode()))) { 
			i = failedBuffer.get(Integer.valueOf(line.hashCode()));
			i++;
			if (i > 9) 
				return;
			failedBuffer.put(Integer.valueOf(line.hashCode()), Integer.valueOf(i));
		} else {
			failedBuffer.put(Integer.valueOf(line.hashCode()), Integer.valueOf(i));
		}
		
		final int times = i;
		timer.schedule(new TimerTask() {
			@Override
			public void run() {
				logger.info("failed object %s, try %s times.", object, times);
				if (!LineExchanger.sendToWriter(line)) 
					logger.info("queue is full, failed object don't send to queue.");
			}
		}, (long) (Math.pow(2, i) * 60 * 1000));
	}
	
	public static synchronized void removeFailedObject(Integer hashcode) {
		failedBuffer.remove(hashcode);
	}
}
